/**
 * Copyright (c) 2013-2019 Nikita Koksharov
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;

import org.redisson.api.RFuture;
import org.redisson.api.RLock;
import org.redisson.client.codec.LongCodec;
import org.redisson.client.codec.StringCodec;
import org.redisson.client.protocol.RedisCommands;
import org.redisson.client.protocol.RedisStrictCommand;
import org.redisson.command.CommandAsyncExecutor;
import org.redisson.pubsub.LockPubSub;

/**
 * redisson读锁。读锁将在链接断开时自动释放。
 * 一个读锁可以被多个线程持有。每个线程可以不同的过期时间持有读锁
 *
 * Lock will be removed automatically if client disconnects.
 *
 * @author Nikita Koksharov
 *
 */
public class RedissonReadLock extends RedissonLock implements RLock {

    public RedissonReadLock(CommandAsyncExecutor commandExecutor, String name) {
        super(commandExecutor, name);
    }

    /**
     * 订阅通道名称: redisson_rwlock:lockName。
     * 当读锁被写锁阻塞时订阅。
     *
     * @return
     */
    @Override
    String getChannelName() {
        return prefixName("redisson_rwlock", getName());
    }

    /**
     * 获取对应的写锁的名称
     * ConnectionId:ThreadId:write
     * @param threadId
     * @return
     */
    String getWriteLockName(long threadId) {
        return super.getLockName(threadId) + ":write";
    }

    /**
     * {lockName}:ConnectionId:ThreadId:rwlock_timeout
     */
    String getReadWriteTimeoutNamePrefix(long threadId) {
        return suffixName(getName(), getLockName(threadId)) + ":rwlock_timeout";
    }

    /**
     * 需要讲一下ReadWriteLock的redis存储结构。注意RedissonReadLock和RedissonWriteLock公用一个redis存储结构
     *
     * 1. 一个名为 lockName 的哈希表。
     * key=mode，value=read/write
     * read模式下，将保存多条持有读锁的记录: key=ConnectionId:ThreadId，value=重入次数
     * write模式下，只会保存一条持有写锁的记录: key=ConnectionId:ThreadId:write，value=重入次数
     * lockName哈希表的过期时间永远是持有读锁的所有线程中最大的ttl
     *
     * 2. 在读锁模式下，使用N个键值对记录持有读锁的线程的过期时间(因为lockName哈希表会存在多个读锁线程，每次加入新线程都会更新读写锁过期时间)
     * key={lockName}:ConnectionId:ThreadId:rwlock_timeout:重入次数
     * value=1
     */

    /**
     * 覆写了最终的上锁代码 tryLockInnerAsync()
     * @param leaseTime 锁的过期时间。要么是用户指定的过期时间，要么是看门狗设置的过期时间
     * @param unit      时间单位
     * @param threadId  当前线程Id
     * @param command   EVAL_LONG:将LUA脚本返回的值原封返回。EVAL_NULL_BOOLEAN:LUA脚本返回null则转换为返回true，LUA脚本返回非null则转换为返回false
     * @param <T>       同上。可能是Long可能是Boolean
     * @return
     */
    @Override
    <T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        internalLockLeaseTime = unit.toMillis(leaseTime);
        // KEYS[1] = lockName
        // KEYS[2] = {lockName}:ConnectionId:ThreadId:rwlock_timeout
        // ARGV[1] = 锁过期时间的毫秒格式
        // ARGV[2] = ConnectionId:ThreadId
        // ARGV[3] = ConnectionId:ThreadId:write
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                // 获取lockName哈希表中的mode记录
                "local mode = redis.call('hget', KEYS[1], 'mode'); " +
                // 如果不存在mode记录或者lockName就不存在
                "if (mode == false) then " +
                    // (添加lockName哈希表)向lockName哈希表中添加 mode:read 这条记录
                    "redis.call('hset', KEYS[1], 'mode', 'read'); " +
                    // 向lockName哈希表中添加 ConnectionId:ThreadId - 1 记录该线程
                    "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                    // 添加一条键值对 {lockName}:ConnectionId:ThreadId:rwlock_timeout:1 - 1
                    "redis.call('set', KEYS[2] .. ':1', 1); " +
                    // 为 {lockName}:ConnectionId:ThreadId:rwlock_timeout:1 这个key设置过期时间
                    "redis.call('pexpire', KEYS[2] .. ':1', ARGV[1]); " +
                    // 为 lockName哈希表重置过期时间
                    "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                    // 返回Nil代表加锁成功
                    "return nil; " +
                "end; " +

                // 如果读写锁存在且就是读锁，或者是写锁但是持有写锁的线程是当前线程
                "if (mode == 'read') or (mode == 'write' and redis.call('hexists', KEYS[1], ARGV[3]) == 1) then " +
                    // 向lockName哈希表中自增式添加重入读锁记录 ConnectionId:ThreadId - ?+1
                    "local ind = redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                    // key就是 {lockName}:ConnectionId:ThreadId:rwlock_timeout:ind重入次数
                    "local key = KEYS[2] .. ':' .. ind;" +
                    // 添加一条键值对 {lockName}:ConnectionId:ThreadId:rwlock_timeout:ind重入次数 - 1 并设置过期时间
                    "redis.call('set', key, 1); " +
                    "redis.call('pexpire', key, ARGV[1]); " +
                    // 更新lockName的过期时间，为所有持有读锁的线程中最大的ttl
                    "local remainTime = redis.call('pttl', KEYS[1]); " +
                    "redis.call('pexpire', KEYS[1], math.max(remainTime, ARGV[1])); " +
                    "return nil; " +
                "end;" +

                // 存在写锁时，读锁无法获取。返回写锁的过期时间
                "return redis.call('pttl', KEYS[1]);",
                Arrays.<Object>asList(getName(), getReadWriteTimeoutNamePrefix(threadId)),
                internalLockLeaseTime, getLockName(threadId), getWriteLockName(threadId));
    }

    /**
     * 读锁的解锁
     * @param threadId ThreadId
     * @return
     */
    @Override
    protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        String timeoutPrefix = getReadWriteTimeoutNamePrefix(threadId);
        String keyPrefix = getKeyPrefix(threadId, timeoutPrefix);

        // KEYS[1] = lockName
        // KEYS[2] = redisson_rwlock:lockName
        // KEYS[3] = {lockName}:ConnectionId:ThreadId:rwlock_timeout
        // KEYS[4] = {lockName}
        // ARGV[1] = 0L
        // ARGV[2] = ConnectionId:ThreadId
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "local mode = redis.call('hget', KEYS[1], 'mode'); " +
                "if (mode == false) then " +
                    // 读写锁已经不存在则向订阅通道发送消息通知其他人
                    "redis.call('publish', KEYS[2], ARGV[1]); " +
                    "return 1; " +
                "end; " +

                // 锁存在且自身已不持有读锁，啥也不干
                "local lockExists = redis.call('hexists', KEYS[1], ARGV[2]); " +
                "if (lockExists == 0) then " +
                    "return nil;" +
                "end; " +

                // 仍持有读锁。重入次数原子减1。从lockName哈希表中删除记录。从整个redis中删除ttl记录
                "local counter = redis.call('hincrby', KEYS[1], ARGV[2], -1); " +
                "if (counter == 0) then " +
                    "redis.call('hdel', KEYS[1], ARGV[2]); " +
                "end;" +
                "redis.call('del', KEYS[3] .. ':' .. (counter+1)); " +

                // 若还有其他人持有读写锁，则整个读写锁新的ttl应该是当前所有持有读写锁的线程中最大的ttl
                "if (redis.call('hlen', KEYS[1]) > 1) then " +
                    "local maxRemainTime = -3; " +
                    "local keys = redis.call('hkeys', KEYS[1]); " +
                    "for n, key in ipairs(keys) do " +
                        "counter = tonumber(redis.call('hget', KEYS[1], key)); " +
                        "if type(counter) == 'number' then " +
                            "for i=counter, 1, -1 do " +
                                "local remainTime = redis.call('pttl', KEYS[4] .. ':' .. key .. ':rwlock_timeout:' .. i); " +
                                "maxRemainTime = math.max(remainTime, maxRemainTime);" +
                            "end; " +
                        "end; " +
                    "end; " +

                    "if maxRemainTime > 0 then " +
                        "redis.call('pexpire', KEYS[1], maxRemainTime); " +
                        "return 0; " +
                    "end;" +

                    "if mode == 'write' then " +
                        "return 0;" +
                    "end; " +
                "end; " +

                // 已没有任何人持有读写锁。则删除锁并发送通知
                "redis.call('del', KEYS[1]); " +
                "redis.call('publish', KEYS[2], ARGV[1]); " +
                "return 1; ",
                Arrays.<Object>asList(getName(), getChannelName(), timeoutPrefix, keyPrefix),
                LockPubSub.UNLOCK_MESSAGE, getLockName(threadId));
    }

    protected String getKeyPrefix(long threadId, String timeoutPrefix) {
        return timeoutPrefix.split(":" + getLockName(threadId))[0];
    }

    @Override
    protected RFuture<Boolean> renewExpirationAsync(long threadId) {
        String timeoutPrefix = getReadWriteTimeoutNamePrefix(threadId);
        String keyPrefix = getKeyPrefix(threadId, timeoutPrefix);

        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "local counter = redis.call('hget', KEYS[1], ARGV[2]); " +
                        "if (counter ~= false) then " +
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +

                        "if (redis.call('hlen', KEYS[1]) > 1) then " +
                        "local keys = redis.call('hkeys', KEYS[1]); " +
                        "for n, key in ipairs(keys) do " +
                        "counter = tonumber(redis.call('hget', KEYS[1], key)); " +
                        "if type(counter) == 'number' then " +
                        "for i=counter, 1, -1 do " +
                        "redis.call('pexpire', KEYS[2] .. ':' .. key .. ':rwlock_timeout:' .. i, ARGV[1]); " +
                        "end; " +
                        "end; " +
                        "end; " +
                        "end; " +

                        "return 1; " +
                        "end; " +
                        "return 0;",
                Arrays.<Object>asList(getName(), keyPrefix),
                internalLockLeaseTime, getLockName(threadId));
    }

    @Override
    public Condition newCondition() {
        throw new UnsupportedOperationException();
    }

    @Override
    public RFuture<Boolean> forceUnlockAsync() {
        cancelExpirationRenewal(null);
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "if (redis.call('hget', KEYS[1], 'mode') == 'read') then " +
                        "redis.call('del', KEYS[1]); " +
                        "redis.call('publish', KEYS[2], ARGV[1]); " +
                        "return 1; " +
                        "end; " +
                        "return 0; ",
                Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE);
    }

    @Override
    public boolean isLocked() {
        RFuture<String> future = commandExecutor.writeAsync(getName(), StringCodec.INSTANCE, RedisCommands.HGET, getName(), "mode");
        String res = get(future);
        return "read".equals(res);
    }

}
